package com.shujia.flink.core

import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

object Demo9WindowProcess {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)

    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    val wordsDS: DataStream[String] = linesDS.flatMap(_.split(','))

    //按照单词分组
    val keyByDS: KeyedStream[String, String] = wordsDS.keyBy(word => word)

    //统计最近5秒单词的数量
    val windowDS: WindowedStream[String, String, TimeWindow] = keyByDS
      .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))

    /**
     * 在窗口之后使用process函数，一次传入一个窗口的数据
     *
     */
    val countDS: DataStream[(String, Long, Long, Int)] = windowDS.process(
      new ProcessWindowFunction[String, (String, Long, Long, Int), String, TimeWindow] {
        /**
         * process: 一个窗口处理一次
         *
         * @param key      ： key
         * @param context  ： 上下文对象，可以获取窗口的开始和结束时间
         * @param elements ：窗口内所有的数据
         * @param out      ：用于将数据发送到下游
         */
        override def process(key: String,
                             context: Context,
                             elements: Iterable[String],
                             out: Collector[(String, Long, Long, Int)]): Unit = {
          //计算单词的数量
          val count: Int = elements.size
          //获取窗口的开始和结束时间
          val window: TimeWindow = context.window
          val startTime: Long = window.getStart
          val endTime: Long = window.getEnd

          //将统计的结果发送到下游
          out.collect((key, startTime, endTime, count))
        }
      })

    countDS.print()

    env.execute()

  }

}
